package MapReduce.Demo3_InputFormat;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import utils.JobSubmit;

import java.io.IOException;

/**
 * @Author lixin
 * @Date 2023/3/15 10:56
 */
public class SequenceFileInputFormatApp {

    public static class SFIFMapper extends Mapper<IntWritable,IntWritable,IntWritable, IntWritable>{

        @Override
        protected void map(IntWritable year, IntWritable temperature, Mapper<IntWritable, IntWritable, IntWritable, IntWritable>.Context context) throws IOException, InterruptedException {

            context.write(year,temperature);

        }
    }

    public static class SFIFReducer extends Reducer<IntWritable,IntWritable,IntWritable,IntWritable> {

        @Override
        protected void reduce(IntWritable key, Iterable<IntWritable> values, Reducer<IntWritable, IntWritable, IntWritable, IntWritable>.Context context) throws IOException, InterruptedException {

            int sumTemp = 0;

            for (IntWritable value : values) {
                sumTemp += value.get();
            }

            context.write(key,new IntWritable(sumTemp));

        }
    }


    public static void main(String[] args) throws Exception {

//        Configuration conf = new Configuration();
//
//        Job job = Job.getInstance(conf);
//
//        job.setMapOutputKeyClass(IntWritable.class);
//        job.setMapOutputValueClass(IntWritable.class);
//        job.setOutputKeyClass(IntWritable.class);
//        job.setOutputValueClass(IntWritable.class);
//
//        job.setJarByClass(SequenceFileInputFormatApp.class);
//        job.setMapperClass(SFIFMapper.class);
//        job.setReducerClass(SFIFReducer.class);
//
//
//        Path outputPath = new Path(args[1]);
//
//        FileSystem fs = FileSystem.get(conf);
//
//        if (fs.exists(outputPath)) {
//            fs.delete(outputPath,true);
//        }
//
//        //设置输入格式化的类型，如果不设置，模式任用TextInputFormat
//        job.setInputFormatClass(SequenceFileInputFormat.class);
//
//        FileInputFormat.setInputPaths(job,new Path(args[0]));
//        FileOutputFormat.setOutputPath(job,outputPath);
//
//        job.waitForCompletion(true);

        JobSubmit.submitBaseJob(SequenceFileInputFormatApp.class,args,SequenceFileInputFormat.class);

    }

}
